package main

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"go.uber.org/zap"
	"time"
)

type OrderListener struct{}

func (s *OrderListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
	fmt.Println("开始执行本地")
	// 执行本地逻辑
	time.Sleep(10 * time.Second)
	// 执行成功进行commit
	return primitive.CommitMessageState
}
func (s *OrderListener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
	return primitive.RollbackMessageState
}

func ProduceTX() {

	Producer, err := rocketmq.NewTransactionProducer(&OrderListener{}, producer.WithNameServer([]string{"192.168.56.101:9876"}))
	if err != nil {
		panic(err)
	}

	if err := Producer.Start(); err != nil {
		panic(err)
	}
	msg := primitive.NewMessage("Test", []byte("Send a Transaction Message"))

	ret, err := Producer.SendMessageInTransaction(context.Background(), msg)
	if err != nil {
		panic(err)
	}
	fmt.Println(ret.String())
	time.Sleep(360 * time.Second)
	err = Producer.Shutdown()
}

func Produce() {

	Producer, err := rocketmq.NewProducer(producer.WithNameServer([]string{"192.168.56.101:9876"}))
	if err != nil {
		panic(err)
	}

	if err := Producer.Start(); err != nil {
		panic(err)
	}
	msg := primitive.NewMessage("Test", []byte("hello world"))

	msg.WithDelayTimeLevel(3) // 实现延迟消息

	ret, err := Producer.SendSync(context.Background(), msg)
	if err != nil {
		panic(err)
	}
	fmt.Println(ret.String())

	err = Producer.Shutdown()
}

func Consume() {
	// 推模式的消费者
	Consumer, err := rocketmq.NewPushConsumer(
		consumer.WithNameServer([]string{"192.168.56.101:9876"}),
		consumer.WithGroupName("my_consumer_group"), // 选择
	)
	if err != nil {
		panic(err)
	}

	err = Consumer.Subscribe("Test", consumer.MessageSelector{}, func(ctx context.Context, message ...*primitive.MessageExt) (consumer.ConsumeResult, error) {

		for i := range message {
			fmt.Println(message[i])
			zap.S().Info(message[i])
		}
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {
		panic(err)
	}
	_ = Consumer.Start()

	// 保证主协程不退出
	time.Sleep(100 * time.Second)
	_ = Consumer.Shutdown()

}

func main() {
	//ProduceTX()
	Consume()
}
